package com.bigdata.kafka.cli;

import com.bigdata.kafka.commons.Commons;
import com.bigdata.kafka.consumer.KafkaConsumerExample;
import com.bigdata.kafka.consumer.KafkaFlinkConsumerExample;
import com.bigdata.kafka.consumer.KafkaSparkConsumerExample;
import com.bigdata.kafka.producer.KafkaProducerExample;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class Main {

    //private static String host;
    private static final Logger logger = LogManager.getLogger(Main.class);
    public static void main(String[] args) {
        String EXAMPLE_GOAL = args[0];
//        String EXAMPLE_GOAL = System.getenv("EXAMPLE_GOAL") != null ?
//                System.getenv("EXAMPLE_GOAL") : "producer";
        logger.info("Kafka Topic: {}", Commons.EXAMPLE_KAFKA_TOPIC);
        logger.info("Kafka Server: {}", Commons.EXAMPLE_KAFKA_SERVER);
        logger.info("Zookeeper Server: {}", Commons.EXAMPLE_ZOOKEEPER_SERVER);
        logger.info("GOAL: {}", EXAMPLE_GOAL);

        switch (EXAMPLE_GOAL.toLowerCase()) {
            case "producer":
                KafkaProducerExample.main();
                break;
            case "consumer.kafka":
                KafkaConsumerExample.main();
                break;
            case "consumer.spark":
                KafkaSparkConsumerExample.main();
                break;
            case "consumer.flink":
                KafkaFlinkConsumerExample.main();
                break;
            default:
                logger.error("No valid goal to run.");
                break;
        }
    }
}

